package main

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"gitee.com/kinwyb/appTools/common/socket"
	"gitee.com/kinwyb/appTools/manager"
	"strconv"
	"strings"
	"sync"
	"time"
)

var tcpListenAddress string  //服务监听地址
var httpListenAddress string //http服务地址
var tcpServer *socket.TcpServer

// 待处理的消息
var pendingMutex = sync.Mutex{}
var pending = make(map[uint64]*manager.Msg)

// 客户端连接信息
type agentInfo struct {
	ID      string  `description:"ID"`
	OS      string  `description:"操作系统"`
	OSARCH  string  `description:"架构"`
	Tag     string  `description:"标签"`
	AppName string  `description:"应用名称"`
	Version string  `description:"运行版本"`
	IsStart bool    `description:"是否运行"`
	Client  string  `description:"客户端连接"`
	RunTime string  `description:"运行时间"`
	Mem     float64 `description:"内存使用(M)"`
	Cpu     float64 `description:"cpu使用"`
}

var agentClientMutex = sync.RWMutex{}
var agentClients = make(map[string]*agentInfo)

type agentInfoSlice []agentInfo

func (x agentInfoSlice) Len() int           { return len(x) }
func (x agentInfoSlice) Less(i, j int) bool { return x[i].ID < x[j].ID }
func (x agentInfoSlice) Swap(i, j int)      { x[i], x[j] = x[j], x[i] }

// 初始化服务
func initTcpServer(ctx context.Context) error {
	vv := strings.SplitN(tcpListenAddress, ":", 2)
	port, err := strconv.ParseInt(vv[1], 10, 64)
	if err != nil {
		manager.ServiceLog.WithField("info", tcpListenAddress).WithError(err).Error("tcp监听地址异常")
		return err
	}
	tcpServer, err = socket.NewTcpServer(ctx, &socket.TcpServerConfig{
		Port:          port,
		ServerAddress: vv[0],
		Log:           manager.ServiceLog,
		ErrorHandler:  nil,
		NewClientHandler: func(clientID string) socket.TcpProtocol {
			manager.ServiceLog.WithField("info", clientID).Info("新客户端")
			agentClientMutex.Lock()
			agentClients[clientID] = &agentInfo{Client: clientID}
			agentClientMutex.Unlock()
			return socket.NewProtocol(nil)
		},
		MessageHandler: msgHandler,
		CloseHandler: func() {
			agentClientMutex.Lock()
			agentClients = nil
			agentClientMutex.Unlock()
			for k := range agentClients {
				tcpServer.CloseClient(k)
			}
		},
		ClientCloseHandler: func(clientID string) { //客户端断开删除
			agentClientMutex.Lock()
			delete(agentClients, clientID)
			agentClientMutex.Unlock()
		},
	})
	if err != nil {
		manager.ServiceLog.WithError(err).Panic("tcp服务器创建失败")
	}
	go checkClientState()
	tcpServer.Listen()
	return nil
}

func checkClientState() {
	defer func() {
		if err := recover(); err != nil {
			manager.ServiceLog.WithField("error", err).Error("客户端状态检测线程异常")
			go checkClientState()
		}
	}()
	t := time.NewTicker(30 * time.Second)
	for {
		<-t.C
		clients := tcpServer.GetAllConnIDs()
		for _, client := range clients {
			getProcessStatus(client)         //获取一下客户端的信息
			sendCmdServerExecVersion(client) //通知服务端运行版本
		}
	}
}

// 消息分类
func msgHandler(clientID string, msg []byte) {
	if len(msg) < 9 {
		go registerClient(clientID) //发送基础数据
		return
	}
	msgID := manager.ByteToUInt(msg[0:8])
	pendingMutex.Lock()
	call := pending[msgID]
	delete(pending, msgID)
	pendingMutex.Unlock()
	if call == nil {
		return
	}
	err := call.UnMarshal(msg)
	if err != nil {
		manager.ServiceLog.WithField("info", fmt.Sprintf("%s - %s", clientID, string(msg))).
			WithError(err).
			Error("客户端数据解析异常")
		call.Clean()
		return
	}
	call.Finish()
}

// sendMsg 发送消息
func sendMsg(clientID string, cmd byte, msg []byte, outTime ...int64) ([]byte, error) {
	if tcpServer == nil {
		return nil, errors.New("tcp server not start")
	}
	call := manager.GetMsg()
	call.Cmd = cmd
	call.Data = msg
	msg = call.Marshal()
	if cap(call.Done) < 1 {
		call.Done = make(chan *manager.Msg, 1)
	}
	pendingMutex.Lock()
	pending[call.ID] = call
	pendingMutex.Unlock()
	err := tcpServer.Write(clientID, msg)
	if err != nil {
		manager.ServiceLog.WithField("info", clientID).WithError(err).Error("消息发送失败")
		pendingMutex.Lock()
		delete(pending, call.ID)
		pendingMutex.Unlock()
		call.Clean()                    //发送失败消息清空
		tcpServer.CloseClient(clientID) //关闭客户端
		return nil, err
	}
	timeOut := int64(10) //默认超时10秒
	if len(outTime) > 0 && outTime[0] > 0 {
		timeOut = outTime[0]
	}
	ctx, _ := context.WithTimeout(context.Background(), time.Duration(timeOut)*time.Second) //10秒超时
	select {
	case call = <-call.Done:
		msg = call.Data
		call.Clean()
		return msg, nil
	case <-ctx.Done():
		pendingMutex.Lock()
		delete(pending, call.ID)
		pendingMutex.Unlock()
		call.Clean()
		return nil, errors.New("消息响应超时")
	}
}

// 注册客户端
func registerClient(clientID string) {
	getProcessStatus(clientID)         //先获取一下客户端的信息,主要获取到应用及系统数据
	sendCmdServerHttpAddress(clientID) //通知http服务地址
	sendCmdServerExecVersion(clientID) //通知服务端运行版本
	startProcess(clientID, true)       //运行程序
}

// 获取进程状态
func getProcessStatus(clientID string) { //获取进程状态
	agentClientMutex.RLock()
	agent := agentClients[clientID]
	agentClientMutex.RUnlock()
	if agent == nil {
		tcpServer.CloseClient(clientID)
		return
	}
	resultData, err := sendMsg(clientID, manager.CommandStateProcess, nil)
	if err != nil {
		return
	}
	state := manager.CmdProcessState{}
	err = state.UnMarshal(resultData)
	if err != nil {
		manager.ServiceLog.WithField("info", clientID).WithError(err).Error("客户端状态获取失败")
		return
	}
	agent.Tag = state.Tag
	agent.OS = strings.SplitN(state.Os, "_", 2)[0]
	agent.ID = state.ID
	agent.AppName = state.App
	agent.Version = state.Version
	agent.IsStart = state.IsStart
	agent.OSARCH = state.Os
	agent.RunTime = fmt.Sprintf("%d", state.StartTime)
	agent.Mem = state.Mem
	agent.Cpu = state.Cpu
}

// 往客户端通知http服务地址
func sendCmdServerHttpAddress(clientID string) {
	msg := manager.CmdServerHttpAddress{
		Status:  'Y',
		Address: httpListenAddress,
	}
	_, err := sendMsg(clientID, manager.CommandServerHttpAddress, msg.Marshal())
	if err != nil {
		return
	}
}

// 往客户端通知服务端执行程序版本
func sendCmdServerExecVersion(clientID string) {
	agentClientMutex.RLock()
	agent := agentClients[clientID]
	agentClientMutex.RUnlock()
	if agent == nil || agent.AppName == "" || agent.OS == "" {
		return
	}
	msg := manager.CmdServerExecVersion{
		Status:  'Y',
		Version: getVersion(agent.AppName, agent.OS),
	}
	if msg.Version == "" {
		msg.Status = 'N'
	}
	_, err := sendMsg(clientID, manager.CommandServerExecVersion, msg.Marshal())
	if err != nil {
		return
	}
}

// 运行程序
func startProcess(clientID string, updateState bool) (bool, error) {
	resState, err := sendMsg(clientID, manager.CommandStartProcess, nil)
	if err != nil {
		return false, errors.New("命令发送失败")
	}
	if len(resState) > 0 && resState[0] == 'Y' {
		if updateState {
			getProcessStatus(clientID) //运行成功重新获取一下运行状态
		}
		return true, nil
	} else {
		errMsg := string(resState[1:])
		if errMsg != "" {
			manager.ServiceLog.WithField("info", clientID).WithField("error", errMsg).Error("运行程序失败")
		}
		return false, errors.New(errMsg)
	}
}

// 停止程序
func stopProcess(clientID string, updateState bool) (bool, error) {
	resState, err := sendMsg(clientID, manager.CommandStopProcess, nil)
	if err != nil {
		return false, errors.New("命令发送失败")
	}
	if len(resState) > 0 && resState[0] == 'Y' {
		if updateState {
			getProcessStatus(clientID) //运行成功重新获取一下运行状态
		}
		return true, nil
	} else {
		errMsg := string(resState[1:])
		if errMsg != "" {
			manager.ServiceLog.WithField("info", clientID).WithField("error", errMsg).Error("停止程序失败")
		}
		return false, errors.New(errMsg)
	}
}

// 更新程序
func updateProcess(clientID string) (bool, error) {
	agentClientMutex.RLock()
	agent := agentClients[clientID]
	agentClientMutex.RUnlock()
	if agent == nil || agent.AppName == "" || agent.OS == "" {
		tcpServer.CloseClient(clientID)
		return false, errors.New("客户端信息获取失败")
	}
	_, err := stopProcess(clientID, false)
	if err != nil {
		return false, err
	}
	msg := manager.CmdServerExecVersion{
		Status:  'Y',
		Version: getVersion(agent.AppName, agent.OS),
	}
	if msg.Version == "" {
		msg.Status = 'N'
	}
	resultData, err := sendMsg(clientID, manager.CommandUpdateProcessExec, msg.Marshal())
	if err != nil {
		return false, errors.New("命令发送失败")
	}
	if len(resultData) > 0 && resultData[0] == 'Y' {
		startProcess(clientID, true)
		return true, nil
	} else {
		errMsg := string(resultData[1:])
		if errMsg != "" {
			manager.ServiceLog.WithField("info", clientID).WithField("error", errMsg).Error("替换程序失败")
		}
		return false, errors.New(errMsg)
	}
}

// 客户端发起http请求
func clientHttpRequest(clientID string, req manager.CmdHttpRequest) ([]byte, error) {
	agentClientMutex.RLock()
	agent := agentClients[clientID]
	agentClientMutex.RUnlock()
	if agent == nil || agent.AppName == "" || agent.OS == "" {
		tcpServer.CloseClient(clientID)
		return nil, errors.New("客户端信息获取失败")
	}
	reqBytes, e := json.Marshal(req)
	if e != nil {
		return nil, errors.New("请求参数序列化失败:" + e.Error())
	}
	resState, err := sendMsg(clientID, manager.CommandHttpRequest, reqBytes, 120)
	if err != nil {
		return nil, errors.New("命令发送失败")
	}
	return resState, nil
}
